#pragma once
#include <stdlib.h>
#include <stdio.h>
#include <time.h>
#include <cstring>
#include <string>
#include <amqp.h>
#include <amqp_framing.h>
#include <amqp_tcp_socket.h>
#ifdef WIN32
#include <windows.h>  
#define usleep(x) Sleep(x/1000>1?x/1000:1)
#else
#include <unistd.h>
#endif

#include "single.h"

using std::string;

typedef int(*msgdisposeFun)(void *, amqp_envelope_t);
int bytes_to_string(amqp_bytes_t bytes, string &retmsg);

class CMQClient
{
public:
	CMQClient();
	~CMQClient();
	int init(const string &hostname, int port);
	int connect();

	int declareExchange(const string &exchange, string etype = "fanout");
	int declareQueue(string &qname,
		amqp_boolean_t passive = 0,
		amqp_boolean_t durable = 0,
		amqp_boolean_t exclusive = 0,
		amqp_boolean_t auto_delete = 0
		);
	int queueBind(const string &qname, const string &exchange, const string &bindingkey);
	int doMsgAck(amqp_envelope_t envelope);

	int getSockfd();
	amqp_connection_state_t getConn();

	int send(const string &exchange, const string &qname, const string &msg);
	int rpc_server_send(const string &exchange, const string &msg, amqp_envelope_t envelope);
	int rpc_client_send(const string &exchange, const string &queue_name,
		const string &msg, const string &reply_to_queue, const string &correlation_id);

	int recv(msgdisposeFun callback, const string &qname, int msg_timeout = 0);
	int rpc_client_recv(const string &reply_to_queue, const string &correlation_id,
		string &rcvdata, int msg_timeout = 0);
	
	
private:
	bool doLogin();
	bool openChannel();
	void clean();
	int dispose_recv_error(int tstart, int msg_timeout, amqp_rpc_reply_t ret);
	void basic_consume(const string& qname);
	int sendBase(const string &exchange, const string &qname,
		const string &msg, amqp_basic_properties_t &props);
	void die_on_amqp_error(amqp_rpc_reply_t x, char const *context);
private:
	string hostname="localhost";
	int port=5672;
	int status=0;
	amqp_socket_t *socket=nullptr;
	amqp_connection_state_t conn;
	amqp_channel_t channel=1;
	bool bLogin=false;
	bool bChannelOpend=false;
	struct timeval consume_timeout;
	string cur_consume_queue="ctp";
	amqp_bytes_t cur_consumer_tag;
};


#define MQ singleton_t<CMQClient>::instance()